package qc.module.qms.dps.consumer;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import qc.common.core.constants.QmsMQConstant;
import qc.common.core.exception.QCPromptException;
import qc.module.platform.dto.db.DBConnectionDto;
import qc.module.qms.dps.RestTemplatePlatform;
import qc.module.qms.dto.mq.QmsMqRawDataMessageDto;
import qc.module.qms.dto.rawdata.QmsRawDataOperateResultDto;
import qc.module.qms.service.rawdata.QmsRawDataRecordService;

import java.util.HashMap;
import java.util.Map;

/**
 * 原始数据保存至数据库Consumer
 *
 * @author QuCheng Tech
 * @since 2025/8/12
 */
@Component
@RocketMQMessageListener(
        topic = QmsMQConstant.MQ_RAWDATA_TOPIC,
        consumerGroup = QmsMQConstant.MQ_RAWDATA_CONSUMER_SAVETODB_GROUP
)
public class RawDataSaveToDBConsumer implements RocketMQListener<QmsMqRawDataMessageDto> {
    private static final Logger logger = LoggerFactory.getLogger(RawDataSaveToDBConsumer.class);

    //模块名称，用于在日志或显示中统一使用
    private static final String moduleName = "原始采集数据消息保存原始数据至数据库";

    @Autowired
    private QmsRawDataRecordService rawDataRecordService;

    @Autowired
    private RestTemplatePlatform restTemplatePlatform;

    /**
     * 维护一个数据库表编码和数据库连接信息对应关系集合，如果接收到的原始数据库表编码没有数据库连接信息调用platform中的接口获取
     */
    private static HashMap<String, DBConnectionDto> tableCodeConnectionMap;

    @Override
    public void onMessage(QmsMqRawDataMessageDto message) {
        logger.debug("onMessage " + moduleName);

        //接收到的消息有效性验证
        String validMessageResult = null;
        if (message == null) {
            validMessageResult = "原始数据消息对象为空";
        } else {
            if (StringUtils.isBlank(message.getTablecode())) {
                validMessageResult = "原始数据消息中数据库表编码为空";
            } else {
                if (message.getDatas() == null || message.getDatas().size() < 0x1) {
                    validMessageResult = "原始数据消息中数据集合为空或数据数量为0";
                }
            }
        }
        //判断如果接收到的消息有效性验证未通过，进行错误日志记录
        if (StringUtils.isNotBlank(validMessageResult)) {
            logger.error("onMessage " + moduleName + "，" + validMessageResult);
            return;
        }

        //一条消息是按数据库表进行划分好的数据，一条消息是同一个数据库表中的数据，一条消息中有多条数据，每条数据中可以包含多个采集要素值
        String tableCode = message.getTablecode();
        logger.debug("onMessage " + moduleName + "，数据库表[" + tableCode + "]，原始数据条数[" + message.getDatas().size() + "]");

        //根据数据库表编码获取对应的数据库连接信息
        DBConnectionDto dbConnection = null;
        try {
            dbConnection = this.getTableConnection(tableCode);

        } catch (QCPromptException e) {
            logger.error("onMessage " + moduleName + "，数据库表[" + tableCode + "]获取对应数据库连接信息异常，" + e.getMessage());
        }

        if (dbConnection == null) {
            //没有对应的数据库连接信息时无法进行保存，进行错误日志记录
            logger.error("onMessage " + moduleName + "，数据库表[" + tableCode + "]，获取对应数据库连接信息为空，无法进行保存原始数据至数据库。");
        } else {
            //调用QmsRawDataRecordService中的save方法保存至原始数据库表，指定发送至MQ的标记为true
            QmsRawDataOperateResultDto saveToDbResult = rawDataRecordService.save(tableCode, message.getDatas(), true, dbConnection);
            if (saveToDbResult != null && saveToDbResult.isSuccess() == true) {
                //保存成功
                logger.info("onMessage " + moduleName + "，数据库表[" + tableCode + "]，原始数据条数[" + message.getDatas().size() + "]，保存至数据库成功。");
            } else {
                //保存到数据库失败
                logger.error("onMessage " + moduleName + "，数据库表[" + tableCode + "]，原始数据条数[" + message.getDatas().size() +
                        "]，保存至数据库失败，提示信息[" + saveToDbResult.getMsg() + "]。");
                //获取失败的集合，在日志中显示失败信息；此处不需要重复记录日志，在rawDataRecordService保存方法中已经进行逐条失败的日志记录
                Map<String, String> fails = saveToDbResult.getFails();
                if (fails != null && fails.size() > 0x0) {

                }
            }
        }
    }

    /**
     * 根据指定的原始数据表编码获取对应的数据库连接；如果在保存的数据库连接Map中已有对应返回已有，如果没有调用platform接口获取后缓存到Map中
     *
     * @param tableCode 原始数据表编码
     * @return qc.module.platform.dto.db.DBConnectionDto
     * @author QuCheng Tech
     * @since 2025/11/1
     */
    DBConnectionDto getTableConnection(String tableCode) throws QCPromptException {
        if (StringUtils.isNotBlank(tableCode)) {
            if (tableCodeConnectionMap == null)
                tableCodeConnectionMap = new HashMap<>();

            //如果根据key存在已有的数据库连接，直接返回
            if (tableCodeConnectionMap.size() > 0x0 && tableCodeConnectionMap.containsKey(tableCode))
                return tableCodeConnectionMap.get(tableCode);

            //根据key获取不到已有的数据库连接，调用platform获取数据库连接信息
            DBConnectionDto tableConnection = restTemplatePlatform.getDbTableConnectionInfo(tableCode);
            if (tableConnection != null) {
                tableCodeConnectionMap.put(tableCode, tableConnection);

                return tableConnection;
            }
        }

        return null;
    }
}